package com.example.canal.service.kafka;

import com.example.canal.event.EventInfo;
import com.example.canal.model.SyncDataRowItemDTO;
import com.example.canal.redis.util.JsonUtils;
import com.example.canal.service.IEventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

@Service
public class KafkaHandler implements IEventHandler {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaHandler.class);

    @Autowired
    private KafkaTemplate<String,Object> syncKafkaTemplate;

    @Override
    public void handle(EventInfo rowData) {
        LOGGER.info("收到数据表变更消息：rowData={}",rowData);
        ListenableFuture<SendResult<String,Object>> future = this.syncKafkaTemplate.send("canal_row_data_change",
                JsonUtils.objectToString(SyncDataRowItemDTO.buildFromRowData(rowData)));
        future.addCallback(new SuccessCallback<SendResult<String, Object>>() {
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                LOGGER.info("消息发送成功：result={}",JsonUtils.objectToString(result));
            }
        }, new FailureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                LOGGER.error("消息发送失败:",ex);
            }
        });
    }
}
